OpenAIを使って、Slackに届くエラー通知に自動応答させてみた

たかやま

2023.03.17

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

こんにちは。たかやまです。

このChatGPT(OpenAI)のビッグウェーブに乗りたいなと思いネタを考えていたところ、ちょうどslackに届く1通のエラーメッセージを目にし、ふとネットで話題になっていた「ちゃんとエラーを読んで」を思い出しました。

エンジニア駆け出しの頃、私もエラーメッセージに苦労したことを思い出しながら、「エラーをOpenAIに読ませたら万事解決じゃね？」と思いを馳せ今回はOpenAIを使ったエラー通知に自動応答するシステムを作ってみました。

全体構成

全体構成はこちらになります。

やってみた

OpenAI API Key

まずは、OpenAIを利用するためのAPI Keyを発行していきます。
OpenAIにアカウント登録済みの方はこちらのリンクからAPI Keyを取得することができます。

https://platform.openai.com/account/api-keys

API Keyはこのあとの手順で行うSSM Parameter Storeに保管するまでメモしておいてください。

Incomming Webhooks

つづいて、OpenAIレスポンスを受け取るためのIncomming Webhooksを作成します。
作成方法はこちらのブログを参考にしてください。

作成したWebhook URLもこのあとSSM Parameter Storeに保管するのでメモしておいてください。

CDK(API Gateway + Lambda)

CDK作成前にさきほど発行したOpenAI API KeyとWebhook URLをSSM Parameter Storeに保管していきます。
パラメータ名はLambda,CDK内で参照しているので、以下のパラメータ名と種類で登録してください。

今回作成するAPI Gateway,Lambdaの設定はこちらになります。

import * as cdk from 'aws-cdk-lib';
import * as apigw from 'aws-cdk-lib/aws-apigateway';
import * as iam from 'aws-cdk-lib/aws-iam';
import * as lambda from 'aws-cdk-lib/aws-lambda';
import * as ssm from 'aws-cdk-lib/aws-ssm';
import { Construct } from 'constructs';
import * as path from 'path';

export class CdkSlackOpenaiStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);
    /**
     * Create Lambda function
     */
    // Read ParameterStore
    const webhookUrl = ssm.StringParameter.fromStringParameterAttributes(this, 'WebhookUrl', {
      parameterName: '/openai/webhook-url',
    });

    // OpenAI Lambda Layer
    const openaiLayer = new lambda.LayerVersion(this, 'OpenaiLayer', {
      layerVersionName: 'openai',
      code: lambda.Code.fromAsset(path.join(__dirname, './assets/openai-layer')),
      compatibleRuntimes: [lambda.Runtime.PYTHON_3_9],
      description: 'OpenAI Lambda Layer',
    });

    // Lambda Function
    const lambdaFunction = new lambda.Function(this, 'OpenaiLambda', {
      code: lambda.Code.fromAsset(path.join(__dirname, './assets')),
      handler: 'lambda_function.lambda_handler',
      runtime: lambda.Runtime.PYTHON_3_9,
      environment: { WEBHOOK_URL: webhookUrl.stringValue },
      functionName: 'openai-lambda',
      layers: [openaiLayer],
      timeout: cdk.Duration.minutes(5),
    });
    lambdaFunction.addToRolePolicy(
      new iam.PolicyStatement({
        effect: iam.Effect.ALLOW,
        actions: ['ssm:GetParameter'],
        resources: [`arn:aws:ssm:${cdk.Stack.of(this).region}:${cdk.Stack.of(this).account}:parameter/openai/*`],
      })
    );

    /**
     * Create API Gateway
     */
    const api = new apigw.RestApi(this, 'OpenaiAPIGateway', {
      cloudWatchRole: true,
      deployOptions: {
        loggingLevel: apigw.MethodLoggingLevel.INFO,
      },
      restApiName: 'openai-api-gateway',
    });
    const lambdaIntegration = new apigw.LambdaIntegration(lambdaFunction);
    api.root.addMethod('POST', lambdaIntegration);
  }
}

こちらの実装はローカルフォルダのOpenAIパッケージをLambda Layerに登録する処理を行っているので、このコードを使われる方は事前にlibディレクトリ配下に/assets/openai-layer/pythonフォルダを作成してOpenAIパッケージを同梱してください。

pip install openai -t lib/assets/openai-layer/python/

Lambdaにデプロイされるコードはこちらです。

import json
import os

import boto3
from urllib3 import PoolManager, exceptions

import openai

# パラメータストアから暗号化されたAPI Keyを取得する
ssm_client = boto3.client("ssm")
ssm_response = ssm_client.get_parameter(Name="/openai/api-key", WithDecryption=True)
openai.api_key = ssm_response["Parameter"]["Value"]

# Webhook URLを設定
slack_webhook_url = os.environ["WEBHOOK_URL"]


def lambda_handler(event, context):
    print(event)
    # Slack Event Subscriptionsのチャレンジレスポンス用
    if "challenge" in event["body"]:
        body = json.loads(event["body"])
        return {
            "statusCode": 200,
            "headers": {"Content-Type": "application/json"},
            "body": json.dumps({"challenge": body["challenge"]}),
        }

    # body内の投稿内容を取得する(プレーンメッセージ)
    json_body_object = json.loads(event["body"])
    if "event" in json_body_object and "text" in json_body_object["event"] and json_body_object["event"]["text"]:
        input_text = json_body_object["event"]["text"]
    else:
        # attachments内の投稿内容を取得する(AWS Chatbotのような加工されたメッセージ)
        input_text = []
        attachment = json_body_object["event"]["attachments"]
        for block in attachment[0]["blocks"]:
            if "text" in block:
                input_text.append(block["text"]["text"])
        input_text = "\n".join(input_text)

    # OpenAIの回答時、またはSlackの再送処理によるイベントの場合処理を行わない
    if "### OpenAIの回答 ###" in input_text or "X-Slack-Retry-Num" in event["headers"]:
        print("do nothing.")
        return {"statusCode": 200, "headers": {"Content-Type": "application/json"}, "body": "OK"}
    else:
        return_message = generate_openai_response(input_text)
        post_to_slack_webhook(slack_webhook_url, "### OpenAIの回答 ###\n" + return_message)
        print("Sent successfully.")

    return {"statusCode": 200, "headers": {"Content-Type": "application/json"}, "body": "OK"}


def post_to_slack_webhook(webhook_url, message):
    http = PoolManager()
    payload = {"text": message}
    headers = {"Content-type": "application/json"}
    try:
        response = http.request("POST", webhook_url, body=json.dumps(payload).encode("utf-8"), headers=headers)
        if response.status != 200:
            raise ValueError(
                f"Request to slack returned an error {response.status}, the response is:\n{response.data.decode('utf-8')}"
            )
    except exceptions.HTTPError as e:
        raise ValueError(f"Request to slack returned an error: {e}")


def generate_openai_response(content):
    engine = "gpt-3.5-turbo"

    completions = openai.ChatCompletion().create(
        model=engine,
        messages=[
            {
                "role": "system",
                "content": "最初にエラー内容を要約してください。",
            },
            {
                "role": "system",
                "content": "エラー内容が解決しそうであれば、解決策を提示してください。解決策の提示が難しい場合は追加情報を求めない形でネクストアクションを提示してください。",
            },
            {
                "role": "system",
                "content": "やり取りは一回で終わらせるようにしてください。",
            },
            {
                "role": "system",
                "content": "なるべく回答は箇条書きにしてください。",
            },
            {
                "role": "system",
                "content": "回答は日本語で返してください。",
            },
            {
                "role": "user",
                "content": content,
            },
        ],
    )

    return completions.choices[0].message.content

Event Subscriptions

API Gateway, Lambdaのデプロイが完了したらSlackからAPI Gatewayにメッセージを転送するためにEvent Subscriptionsの設定をしていきます。

はじめにEvent SubscriptionsのChallenge認証を行っていきます。

Incomming Webhooksを作成したslack appsでEvent Subscriptionsを選択してください。
Enable EventsがOffの場合はOnにしてください。

OnにするとRequest URLが表示されるので、最初に作成したAPI GatewayのURLをこちらに入力していきます。

API GatewayのURLはStages -> Invoke URLから確認することができます。

URLをメモできたら、さきほどのRequest URLに入力していきます。
URL入力後Challenge認証が走り、無事認証が通ればVerifiedが表示されます。

認証が通ったあとはSlackボットがSlackにアクセスできる権限を設定していきます。
ここではチャンネルにメッセージが投稿されたらイベント通知したいので以下のmessage.channelsを追加します。

追加できたら、Save Changesを選択します。

slack apiの設定が完了したら、Incomming Webhooksを設定しているチャンネルにEvent Subscriptionsを設定したアプリを追加していきます。
ここでは#notify-alarmというチャンネルを用意しアプリを追加していきます。追加するチャンネルのドロップダウンボタンを選択します。

ポップアップが表示されたら、インテグレーション -> アプリを追加するを選択します。

作成したアプリ(ここではOpenAI)が表示されているので、追加を選択します。

Incomming Webhooksとアプリ(Event Subscriptions)追加のメッセージが表示されていれば準備完了です！

自動応答を確認してみる

GuardDuty

では試しにGuardDutyのテストイベントを発生させて、AWS Chatbot経由で通知してみたいと思います。

こちらのテストコマンドでBackdoor:EC2/DenialOfService.UnusualProtocolを発生させてみます。

aws guardduty create-sample-findings \
--detector-id $(aws guardduty list-detectors --query 'DetectorIds' --output text) \
--finding-types Backdoor:EC2/DenialOfService.UnusualProtocol